/*
 * To change this template, choose Tools | Templates
 * and open the template in the editor.
 */
package com.meego.common.thrift;

import com.meego.common.config.Config;
import com.meego.common.log.GA;
import java.util.Map;
import java.util.concurrent.ArrayBlockingQueue;
import org.apache.thrift.TServiceClientFactory;
import org.apache.thrift.protocol.TBinaryProtocol;
import org.apache.thrift.protocol.TProtocol;
import org.apache.thrift.transport.TFramedTransport;
import org.apache.thrift.transport.TSocket;
import org.apache.thrift.transport.TTransport;
import org.cliffc.high_scale_lib.NonBlockingHashMap;

/**
 *
 * @author TueHM
 */
public class TClientManager {

    private static Map<String, TClientManager> instances = new NonBlockingHashMap();
    private ArrayBlockingQueue<TServiceBase.Client> queue;
    private TServiceClientFactory factory;
    private int maxPool;
    private int minPool;
    private String host;
    private Integer port;

    public static TClientManager getInstance(String name) {
        TClientManager instance = (TClientManager) instances.get(name);
        if (instance == null) {
            GA.service.error("Call function setup before getInstance(String name)");
        }
        return instance;
    }

    public static void setup(String name, TServiceClientFactory factory) {
        TClientManager instance = (TClientManager) instances.get(name);

        if (instance != null) {
            return;
        }

        synchronized (TClientManager.class) {
            if (instance == null) {
                instance = new TClientManager(name, factory);
                instances.put(name, instance);
            }
        }
    }

    public TClientManager(String name, TServiceClientFactory factory) {
        this.host = Config.getStrConfig(String.format("%s.%s", name, "host"), "0.0.0.0");
        this.port = Config.getIntConfig(String.format("%s.%s", name, "port"), 0);

        this.maxPool = Config.getIntConfig(String.format("%s.%s", name, "maxpool"), 1024);
        this.minPool = Config.getIntConfig(String.format("%s.%s", name, "minpool"), 128);

        this.queue = new ArrayBlockingQueue(this.maxPool);

        this.factory = factory;
    }

    public TServiceBase.Client borrowClient() {
        TServiceBase.Client client = null;
        if (this.queue.size() > 0) {
            synchronized (TClientManager.class) {
                try {
                    client = (TServiceBase.Client) this.queue.take();
                } catch (InterruptedException ex) {
                    GA.service.error(ex);
                }
            }
        }

        if (client == null || !isAlive(client)) {
            TTransport transport = new TFramedTransport(new TSocket(this.host, this.port.intValue()));
            TProtocol protocol = new TBinaryProtocol(transport);
            try {
                transport.open();
                client = (TServiceBase.Client) this.factory.getClient(protocol);
            } catch (Exception ex) {
                GA.service.error(ex);
            }

        }

        return client;
    }

    public void returnClient(TServiceBase.Client client) {
        try {
            if (client == null) {
                return;
            }

            if (this.queue.size() >= this.maxPool) {
                destroyClient(client);
                return;
            }

            if ((this.queue.size() > this.minPool) && (!isAlive(client))) {
                destroyClient(client);
                return;
            }
            this.queue.put(client);
        } catch (InterruptedException e) {
            GA.service.error("Exception in put", e);
        }
    }

    public void invalidClient(TServiceBase.Client client) {
        if (client != null) {
            try {
                if (!client.ping()) {
                    destroyClient(client);
                } else {
                    returnClient(client);
                }
            } catch (Exception ex) {
                destroyClient(client);
            }
        }
    }

    private void destroyClient(TServiceBase.Client client) {
        if (client != null) {
            TTransport itrans = client.getInputProtocol().getTransport();
            TTransport otrans = client.getOutputProtocol().getTransport();
            if (itrans != null) {
                itrans.close();
            }
            if ((otrans != null) && (otrans != itrans)) {
                otrans.close();
            }
        }
    }

    private boolean isAlive(TServiceBase.Client client) {
        if (client == null) {
            return false;
        }

        TFramedTransport transport = (TFramedTransport) client.getOutputProtocol().getTransport();

        return (transport != null) && (transport.isOpen());
    }
}
